package com.sparrow.rpc.service;

import com.sparrow.common.thread.ThreadPool;
import com.sparrow.common.thread.factory.SimpleThreadFactory;
import com.sparrow.rpc.domain.RpcNode;
import com.sparrow.rpc.future.SettableRpcFuture;
import com.sparrow.rpc.manager.RpcClientManager;
import com.sparrow.rpc.manager.RpcNodeManager;
import com.sparrow.rpc.socket.RpcClient;
import com.sparrow.rpc.annotation.RpcInterface;
import com.sparrow.rpc.annotation.RpcMethod;
import com.sparrow.rpc.domain.RpcWrapper;
import com.sparrow.rpc.future.RpcFuture;
import com.sparrow.rpc.socket.msg.req.RpcRequest;
import com.sparrow.rpc.socket.msg.resp.RpcResponse;
import com.sparrow.rpc.socket.session.RpcSession;
import com.sparrow.rpc.util.RpcUtil;
import io.netty.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Rpc 服务类
 */
@Component
public class RpcService implements DisposableBean, BeanPostProcessor {
    private static final Logger logger = LoggerFactory.getLogger(RpcService.class);

    // 存储请求 ID 与 RpcFuture 的映射关系
    private static final Map<Long, RpcFuture> futureMap = new ConcurrentHashMap<>();
    // 存储方法键与 RpcWrapper 的映射关系，用于快速查找方法对应的封装信息
    private final Map<String, RpcWrapper> wrapperMap = new ConcurrentHashMap<>();
    /**
     * 单线程执行
     */
    private final ExecutorService executorService = Executors.newSingleThreadExecutor(new SimpleThreadFactory("RPC"));
    // 单例实例
    public static RpcService INSTANCE;
    // 标记服务是否已销毁
    private final AtomicBoolean isDestroyed = new AtomicBoolean(false);
    
    @Autowired
    private RpcClientManager rpcClientManager;
    @Autowired
    RpcNodeManager rpcNodeManager;
    /**
     * 构造函数，初始化单例实例
     */
    public RpcService() {
        INSTANCE = this;
    }

    /**
     * 获取 RpcService 的单例实例
     *
     * @return RpcService 单例实例
     */
    public static RpcService getInstance() {
        return INSTANCE;
    }

    /**
     * Spring Bean 后置处理器方法，在 Bean 初始化后执行。
     * 扫描 Bean 实现的接口，查找带有 @RpcInterface 注解的接口，并注册带有 @RpcMethod 注解的方法。
     *
     * @param bean     初始化后的 Bean 实例
     * @param beanName Bean 的名称
     * @return 处理后的 Bean 实例
     * @throws BeansException 如果在处理过程中发生 Spring Bean 相关异常
     */
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (isDestroyed.get()) {
            return bean;
        }
        // 遍历 Bean 实现的所有接口
        for (Class<?> interfaceClass : bean.getClass().getInterfaces()) {
            // 检查接口是否带有 @RpcInterface 注解
            if (!interfaceClass.isAnnotationPresent(RpcInterface.class)) {
                continue;
            }
            // 遍历接口中的所有方法
            for (Method method : interfaceClass.getDeclaredMethods()) {
                // 检查方法是否带有 @RpcMethod 注解
                if (!method.isAnnotationPresent(RpcMethod.class)) {
                    continue;
                }
                // 检查方法的返回类型是否符合 RPC 要求
                if (RpcUtil.checkRpcReturnType(method)) {
                    register(bean, method);
                }
            }
        }
        return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
    }

    /**
     * 注册 RPC 方法，将方法信息封装到 RpcWrapper 中，并存储到 wrapperMap 中。
     *
     * @param object 方法所属的对象实例
     * @param method 要注册的方法
     */
    private void register(Object object, Method method) {
        if (isDestroyed.get()) {
            return;
        }
        // 生成方法的唯一键
        String methodKey = RpcUtil.getMethodKey(method);
        RpcWrapper rpcWrapper = new RpcWrapper(method, object, methodKey);
        // 将 RpcWrapper 存储到 wrapperMap 中，如果键已存在则抛出异常
        RpcWrapper previousWrapper = wrapperMap.putIfAbsent(methodKey, rpcWrapper);
        if (previousWrapper != null) {
            throw new RuntimeException("Rpc 方法重复定义 info: " + methodKey);
        }
    }


    /**
     * 根据节点 ID 和接口类型获取代理对象。
     *
     * @param nodeId 节点 ID
     * @param clazz  接口类型
     * @param <T>    接口类型的泛型
     * @return 代理对象
     */
    public <T> T getProxy(String nodeId, Class<T> clazz) {
        if (isDestroyed.get()) {
            return null;
        }
        RpcClient client = rpcClientManager.getRpcClient(nodeId);
        if (client == null) {
            return null;
        }
        return client.getProxy(clazz);
    }

    /**
     * 根据方法键获取 RpcWrapper 实例。
     *
     * @param methodKey 方法键
     * @return RpcWrapper 实例，如果不存在则返回 null
     */
    public RpcWrapper getRpcWrapper(String methodKey) {
        if (isDestroyed.get()) {
            return null;
        }
        return wrapperMap.get(methodKey);
    }

    /**
     * 处理 RPC 请求，在单线程执行器中异步执行。
     *
     * @param session    RPC 会话
     * @param rpcRequest RPC 请求
     */
    public void request(RpcSession session, RpcRequest rpcRequest) {
        if (isDestroyed.get()) {
            return;
        }
        executorService.execute(() -> {
            // 根据请求的方法键查找 RpcWrapper
            RpcWrapper rpcWrapper = getRpcWrapper(rpcRequest.getMethodKey());
            if (rpcWrapper == null) {
                // 若未找到方法封装信息，返回错误响应
                respWithError(session, rpcRequest, new RuntimeException("not exist method: " + rpcRequest.getMethodKey()));
                return;
            }
            try {
                // 创建 RPC 响应对象
                RpcResponse rpcResponse = new RpcResponse();
                rpcResponse.setUuid(rpcRequest.getUuid());
                // 调用方法并获取结果
                Object invokeResult = rpcWrapper.getMethod().invoke(rpcWrapper.getObject(), rpcRequest.getArgs());
                // 检查方法返回类型是否为非 void
                if (!RpcUtil.isVoidMethod(rpcWrapper.getMethod().getReturnType())) {
                    if (invokeResult instanceof SettableRpcFuture<?>) {
                        SettableRpcFuture<?> settableRpcFuture = (SettableRpcFuture<?>) invokeResult;
                        settableRpcFuture.addListener((result, throwable) -> {
                            if (throwable != null) {
                                respWithError(session, rpcRequest, throwable);
                            } else {
                                rpcResponse.setResult(result);
                                session.writeAndFlush(rpcResponse);
                            }
                        });
                    } else {
                        // 获取 RpcFuture 中的结果
                        Object result = ((RpcFuture<?>) invokeResult).getResult();
                        rpcResponse.setResult(result);
                        // 发送响应
                        session.writeAndFlush(rpcResponse);
                    }
                }
            } catch (Throwable e) {
                // 若出现异常，返回错误响应
                respWithError(session, rpcRequest, e);
            }
        });
    }

    /**
     * 发送错误响应，将错误信息封装到 RpcResponse 中并发送给客户端。
     *
     * @param session    RPC 会话
     * @param rpcRequest RPC 请求
     * @param error      错误信息
     */
    public void respWithError(RpcSession session, RpcRequest rpcRequest, Throwable error) {
        if (isDestroyed.get()) {
            return;
        }
        RpcResponse rpcResponse = new RpcResponse();
        rpcResponse.setUuid(rpcRequest.getUuid());
        rpcResponse.setError(RpcUtil.getErrorInfo(error));
        session.writeAndFlush(rpcResponse);
    }

    /**
     * 处理 RPC 响应，在单线程执行器中异步执行。
     * 从 futureMap 中移除对应的 RpcFuture 并触发响应处理。
     *
     * @param response RPC 响应
     */
    public void response(RpcResponse response) {
        if (isDestroyed.get()) {
            return;
        }
        executorService.execute(() -> {
            // 从 futureMap 中移除对应的 RpcFuture
            RpcFuture future = futureMap.remove(response.getUuid());
            if (future == null) {
                return;
            }
            // 触发 RpcFuture 的响应处理
            future.complete(response);
        });
    }

    /**
     * 添加 RpcFuture 到 futureMap 中，并设置超时任务。
     *
     * @param future RpcFuture 实例
     */
    public static void addFuture(RpcFuture future) {
        futureMap.put(future.getRequestId(), future);
        addTimeOut(future);
    }

    /**
     * 为 RpcFuture 添加超时任务，若超时则移除 RpcFuture 并触发超时处理。
     *
     * @param future RpcFuture 实例
     */
    public static void addTimeOut(RpcFuture future) {
        // 取消之前的超时任务
        future.cancelTimeOutTask();
        // 创建新的超时任务
        Timeout timeout = ThreadPool.TIMER.newTimeout(t -> {
            removeFuture(future);
            future.onTimeOut();
        }, future.getTimeOutSec(), TimeUnit.SECONDS);
        future.setTimeOutTask(timeout);
    }

    /**
     * 从 futureMap 中移除指定的 RpcFuture。
     *
     * @param future RpcFuture 实例
     */
    public static void removeFuture(RpcFuture future) {
        futureMap.remove(future.getRequestId());
    }

    /**
     * Spring 容器销毁时调用的方法，停止定时器和执行器。
     *
     * @throws Exception 如果在销毁过程中发生异常
     */
    @Override
    public void destroy() throws Exception {
        if (isDestroyed.compareAndSet(false, true)) {
            // 关闭执行器
            executorService.shutdown();
            // 等待执行器中的任务完成，最多等待 20 分钟
            if (!executorService.awaitTermination(20, TimeUnit.MINUTES)) {
                executorService.shutdownNow();
            }
            logger.info("RpcService关闭");
        }
    }

    public RpcNode getNode(String nodeId) {
        return rpcNodeManager.getNode(nodeId);
    }
}